package realtime.topology;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

import java.util.Map;

/**
 * Created by luoqifei on 17-5-3.
 */
public class RedisTopology {
    private final static String ID_KAFKA_SPOUT = "redis-read-kafka-spout1";
    private final static String ID_REDIS_BOLT = "avro-save2redis-bolt1";
    private Map<String, String> env;
    public RedisTopology() {
        env = System.getenv();
    }
    public static void main(String[] args) {

        int redisBoltTasks=1;
        int spountTasks = 1;
        Config config = new Config();
        config.setDebug(false);
        LocalCluster cluster = new LocalCluster();
        config.setNumWorkers(1);
        config.setMaxTaskParallelism(redisBoltTasks+1);
        // create topology
        RedisTopology topology = new RedisTopology();
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout(ID_KAFKA_SPOUT, new RedisSpout(),spountTasks);

        //builder.setBolt(ID_REDIS_BOLT, topology.getRedisAvroBolt(),
                //redisBoltTasks).noneGrouping(ID_KAFKA_SPOUT);

        builder.setBolt(ID_REDIS_BOLT, topology.getSaveByPointsBolt(),
                redisBoltTasks).noneGrouping(ID_KAFKA_SPOUT);

        config.setMaxSpoutPending(500);
        config.setMaxTaskParallelism(redisBoltTasks+spountTasks);
        // submit topology locally
        cluster.submitTopology(RedisTopology.class.getSimpleName(), config, builder.createTopology());
        while (true) ;

    }
    public RedisBolt getRedisAvroBolt() {
        RedisBolt bolt = null;
        try {
            bolt = new RedisBolt();
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
        return bolt;
    }
    public SaveByPointsBolt getSaveByPointsBolt() {
        SaveByPointsBolt bolt = null;
        try {
            bolt = new SaveByPointsBolt();
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        }
        return bolt;
    }
}